package com.cfh.binder.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.List;

public class RocketMQMessagChannelBinder implements
        Binder<MessageChannel, ConsumerProperties, ProducerProperties> {

    private static final String TOPIC = "TEST_TOPIC";
    private static final String TAG = "TEST_TAG";
    private static final String GROUP = "TEST_GROUP";
    private static final String NAME_ADRESS = "localhost:9876";

    @Override
    public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inputChannel,
                                                ConsumerProperties consumerProperties) {
        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");

        // Specify name server addresses.
        consumer.setNamesrvAddr(NAME_ADRESS);

        // Subscribe one more more topics to consume.
        try {
            consumer.subscribe(TOPIC, TAG);

            // Register callback to execute on arrival of messages fetched from brokers.
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    System.out.printf("消息接收: %s%n", msgs);
                    msgs.forEach(msg -> {
                        byte[] body = msg.getBody();
                        inputChannel.send(new GenericMessage<>(body));
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            //Launch the consumer instance.
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return () -> {
            System.out.println("consumer shutdowning");
            consumer.shutdown();
        };
    }

    @Override
    public Binding<MessageChannel> bindProducer(String name, MessageChannel outputChannel,
                                                ProducerProperties producerProperties) {
        DefaultMQProducer producer = new
                DefaultMQProducer(GROUP);
        // Specify name server addresses.
        producer.setNamesrvAddr(NAME_ADRESS);
        try {
            //Launch the instance.
            producer.start();
            SubscribableChannel subscribableChannel = (SubscribableChannel) outputChannel;
            subscribableChannel.subscribe(message -> {
                try {
                    Message msg = new Message(TOPIC, TAG,
                            (byte[]) message.getPayload());
                    //Call send message to deliver message to one of brokers.
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("消息发送：%s%n", sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return () -> {
            System.out.println("producer shutdowning");
            producer.shutdown();
        };
    }
}
